package dfvs

import (
	"context"
	"encoding/json"
	"fmt"
	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/sirupsen/logrus"
	"github.com/zing-dev/general-protocol-lib/log"
	"math"
	"sync"
	"time"
)

type App struct {
	ctx    context.Context
	cancel context.CancelFunc
	config *Config

	client mqtt.Client

	Channel     sync.Map
	FiberAlarm  chan MQTTChannelAlarm
	FiberStatus chan MQTTChannelFiber
	FiberSignal chan MQTTChannelSignal
	Heartbeat   chan MQTTHeartbeat
	Message     chan Message
}

func NewApp(ctx context.Context, config *Config) *App {
	ctx, cancel := context.WithCancel(ctx)
	if config.MQTTServerURL == "" {
		config.MQTTServerURL = defaultMQTTServerURL
	}

	if config.AlarmPrecision == float32(0) {
		config.AlarmPrecision = defaultAlarmPrecision
	}

	if config.Debug {
		log.L.Info("---------------config---------------")
		log.L.Info(fmt.Sprintf("MQTT Server: %s", config.MQTTServerURL))
		log.L.Info(fmt.Sprintf("Alarm Precision: %.3f", config.AlarmPrecision))
		log.L.Info(fmt.Sprintf("Alarm Level: %s", config.AlarmLevel))
		log.L.Info(fmt.Sprintf("Alarm Interval: %s", config.AlarmInterval))
		log.L.Info("---------------config---------------")
	}

	return &App{
		ctx:       ctx,
		cancel:    cancel,
		config:    config,
		Channel:   sync.Map{},
		Heartbeat: make(chan MQTTHeartbeat, 0),
		Message:   make(chan Message, 10),
	}
}

func (a *App) setMessage(level logrus.Level, message string) {
	select {
	case a.Message <- Message{
		Level:   level,
		Message: message,
		At:      time.Now(),
	}:
	default:
	}
}

func (a *App) Run() error {
	options := mqtt.NewClientOptions().
		SetClientID(fmt.Sprintf("%s", a.config.MQTTServerURL)).
		AddBroker(a.config.MQTTServerURL).
		SetAutoReconnect(true).
		SetConnectRetry(true).
		SetMaxReconnectInterval(time.Hour).
		SetUsername(a.config.MQTTUsername).
		SetPassword(a.config.MQTTPassword)
	a.client = mqtt.NewClient(options)
	if token := a.client.Connect(); token.Wait() && token.Error() != nil {
		a.setMessage(logrus.FatalLevel, token.Error().Error())
		return token.Error()
	}

	log.L.Info("连接MQTT服务端成功")
	if a.config.TopicFiberAlarm != "" {
		a.FiberAlarm = make(chan MQTTChannelAlarm, 10)
		if token := a.client.Subscribe(a.config.TopicFiberAlarm, 2, func(client mqtt.Client, message mqtt.Message) {
			var alarm = MQTTChannelAlarm{}
			err := json.Unmarshal(message.Payload(), &alarm)
			if err != nil {
				a.setMessage(logrus.ErrorLevel, fmt.Sprintf("解析报警消息失败: %s", err))
				return
			}
			alarm.At = formatTime(alarm.LastPushTime)
			value, ok := a.Channel.Load(alarm.Id())
			if !ok {
				a.setMessage(logrus.ErrorLevel, fmt.Sprintf("未知设备%s,通道%d报警", alarm.SensorID, alarm.ChannelID))
				return
			}
			device := value.(*SensorChannel)
			if AlarmLevel(alarm.Level) != device.AlarmLevel {
				a.setMessage(logrus.WarnLevel, fmt.Sprintf("当前设备%s通道%d报警级别%s与订阅报警级别%s不匹配", alarm.SensorID, alarm.ChannelID, AlarmLevel(alarm.Level), device.AlarmLevel))
				return
			}
			alarmPrecision := float32(math.Abs(float64(alarm.CenterPosition - device.latestAlarmLocation)))
			interval := alarm.At.Sub(device.latestAlarmTime)
			if interval <= device.AlarmInterval && alarmPrecision <= device.AlarmPrecision {
				a.setMessage(logrus.WarnLevel, fmt.Sprintf("当前设备%s通道%d报警精度%.3f小于订阅报警精度%.3f,且报警间隔%s小于订阅报警间隔%s", alarm.SensorID, alarm.ChannelID, alarmPrecision, device.AlarmPrecision, interval, device.AlarmInterval))
				return
			}
			device.latestAlarmTime = alarm.At
			device.latestAlarmLocation = alarm.CenterPosition
			select {
			case a.FiberAlarm <- alarm:
			default:
				a.setMessage(logrus.WarnLevel, "报警数量过多,已舍弃")
			}
		}); token.Wait() && token.Error() != nil {
			a.setMessage(logrus.FatalLevel, fmt.Sprintf("订阅报警消息失败: %s", token.Error()))
		}
	}

	if a.config.TopicFiberStatus != "" {
		a.FiberStatus = make(chan MQTTChannelFiber, 10)
		if token := a.client.Subscribe(a.config.TopicFiberStatus, 2, func(client mqtt.Client, message mqtt.Message) {
			var status = MQTTChannelFiber{}
			err := json.Unmarshal(message.Payload(), &status)
			if err != nil {
				a.setMessage(logrus.ErrorLevel, fmt.Sprintf("解析光纤状态消息失败: %s", err))
				return
			}
			status.At = formatTime(status.PushTime)
			status.FiberStatusText = status.FiberStatus.String()
			select {
			case a.FiberStatus <- status:
			default:
			}
		}); token.Wait() && token.Error() != nil {
			a.setMessage(logrus.FatalLevel, fmt.Sprintf("订阅光纤状态消息失败: %s", token.Error()))
		}
	}

	if a.config.TopicFiberSignal != "" {
		a.FiberSignal = make(chan MQTTChannelSignal, 10)
		if token := a.client.Subscribe(a.config.TopicFiberSignal, 2, func(client mqtt.Client, message mqtt.Message) {
			signal, err := ToSignal(message.Payload())
			if err != nil {
				a.setMessage(logrus.ErrorLevel, fmt.Sprintf("解析光纤信号数据失败: %s", err))
				return
			}
			select {
			case a.FiberSignal <- *signal:
			default:
			}
		}); token.Wait() && token.Error() != nil {
			a.setMessage(logrus.FatalLevel, fmt.Sprintf("订阅光纤信号消息失败: %s", token.Error()))
		}
	}

	a.client.Subscribe(defaultTopicHeartbeat, 2, func(client mqtt.Client, message mqtt.Message) {
		var heartbeat = MQTTHeartbeat{}
		err := json.Unmarshal(message.Payload(), &heartbeat)
		if err != nil {
			a.setMessage(logrus.ErrorLevel, fmt.Sprintf("解析心跳包信号数据失败: %s", err))
			return
		}
		select {
		case a.Heartbeat <- heartbeat:
		default:
		}
	})
	return nil
}
